package com.example.sikong.handler.websocket;

import com.alibaba.fastjson.JSON;
import com.example.common.RedisKeyName;
import com.example.config.SiKongProperties;
import com.example.sikong.enums.SiKongMessageType;
import com.example.sikong.service.RedisProvider;
import com.example.sikong.service.SiKongApiClientService;
import com.example.util.SiKongMapDataUtil;
import com.example.websocket.dto.WSMessageResultDTO;
import com.example.websocket.enums.SupportType;
import com.example.websocket.handler.OrderdWebSocketHandler;
import com.example.websocket.util.WSSessionUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;

import java.util.*;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


/**
 * websocket 定时消息处理器
 */
@Slf4j
@Component
public class HangarWebSocketHandler implements OrderdWebSocketHandler, InitializingBean {

    @Autowired
    private RedisProvider redisProvider;
    @Autowired
    private SiKongApiClientService siKongApiClientService;

    @Autowired
    private SiKongProperties siKongProperties;

    private ScheduledThreadPoolExecutor scheduledExecutor;


    @Override
    public boolean support(SupportType supportType, WebSocketSession session, Object message) {
        return SupportType.handle_message.equals(supportType);
    }

    public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {
        TextMessage messages = (TextMessage) message;
        String payload = messages.getPayload();
        handleLiveReportMessages(payload);
    }

    @Override
    public String getType() {
        return SiKongWebSocketHandler.type;
    }

    @Override
    public int getOrder() {
        return 0;
    }


    /**
     * 创建定时器,
     * 1. 定时发送在线数据
     * 2. 定时关闭未上报的旁路推流
     * 3. 定时关闭线下的无人直播
     */
    protected synchronized void initScheduledExecutorService() {
        if (scheduledExecutor == null) {
            scheduledExecutor = new ScheduledThreadPoolExecutor(siKongProperties.getCoreSize());
            // 定时发送在线机库与飞机的绑定关系
            scheduledExecutor.scheduleAtFixedRate(this::sendOnlineHangarMessage, siKongProperties.getInitialDelay(), siKongProperties.getOnlineMessageSendPeriod(), TimeUnit.SECONDS);
            // 定时关闭空闲直播
            scheduledExecutor.scheduleAtFixedRate(this::closeNoReportLive, siKongProperties.getInitialDelay(), siKongProperties.getCloseReportLivePeriod(), TimeUnit.SECONDS);
            // 定时关闭下线的无人机直播
            scheduledExecutor.scheduleAtFixedRate(this::closeOffLineLive, siKongProperties.getInitialDelay(), siKongProperties.getCloseOffLineLivePeriod(), TimeUnit.SECONDS);
        }
    }

    /**
     * 发送在线信息
     */
    protected void sendOnlineHangarMessage() {
        log.info("发送在线信息");
        try {
            // 获取在
            Set<String> keys = redisProvider.getAllKeys(RedisKeyName.SI_KONG_OSD_HANGAR_ONLINE + ":*");
            List<String> orderKeys = new ArrayList<>(keys);
            List<Object> hangars = redisProvider.mGet(orderKeys);
            List<Map<String, Object>> hangerWithDrone = new ArrayList<>();
            for (int i = 0; i < orderKeys.size(); i++) {
                String key = orderKeys.get(i);
                String hangarSn = key.substring(key.lastIndexOf(":") + 1);
                Map<String, Object> hangar = new HashMap<>();
                Map<String, Object> subs = (Map<String, Object>) hangars.get(i);
                hangar.put(hangarSn, subs);
                hangerWithDrone.add(hangar);
            }
            String payload = JSON.toJSONString(new WSMessageResultDTO(SiKongMessageType.hangar_online.name(), hangerWithDrone));
            WSSessionUtil.sendTextMessage(SiKongWebSocketHandler.type, payload);
        } catch (Exception e) {
            log.error("获取机库在线信息失败:", e);
        }
    }

    /**
     * 关闭未上报的直播
     */
    private void closeNoReportLive() {
        log.info("=====================定时关闭未上报的直播");
        Set<String> allKeys = redisProvider.getAllKeys(RedisKeyName.SI_KONG_SN_CACHE + ":*");
        if (allKeys != null && !allKeys.isEmpty()) {
            for (String allKey : allKeys) {
                redisProvider.del(allKey);
            }
        }

        String inLiveKeyPattern = RedisKeyName.SI_KONG_IN_LIVE + ":*";
        Set<String> inLiveKeys = redisProvider.getAllKeys(inLiveKeyPattern);
        if (inLiveKeys.isEmpty()) {
            return;
        }
        inLiveKeys = inLiveKeys.stream().map(ele -> {
            String sn = ele.substring(0, ele.lastIndexOf(":"));
            return sn.substring(sn.lastIndexOf(":") + 1);
        }).collect(Collectors.toSet());
        // 获取所有上报的key
        String pattern = RedisKeyName.SI_KONG_OSD_REPORT_VIEW + ":*";
        List<String> reportKeys = redisProvider.getAllKeys(pattern).stream().map(ele -> ele.substring(ele.lastIndexOf(":") + 1)).collect(Collectors.toList());
        for (String inLiveSn : inLiveKeys) {
            //  没有在上报中的直播关闭
            if (!reportKeys.contains(inLiveSn)) {
                siKongApiClientService.stopLiveByInLiveKey(inLiveSn);
            }
        }
    }

    /**
     * 关闭下线消息
     */
    private void closeOffLineLive() {
        log.info("=====================关闭下线消息");
        String inLiveKeyPattern = RedisKeyName.SI_KONG_IN_LIVE + ":*";
        Set<String> inLiveKeys = redisProvider.getAllKeys(inLiveKeyPattern);
        if (inLiveKeys.isEmpty()) {
            return;
        }
        inLiveKeys = inLiveKeys.stream().map(ele -> {
            String sn = ele.substring(0, ele.lastIndexOf(":"));
            return sn.substring(sn.lastIndexOf(":") + 1);
        }).collect(Collectors.toSet());
        // 获取所有上报的key
        String dronePattern = RedisKeyName.SI_KONG_OSD_DRONE_ONLINE + ":*";
        String hangarPattern = RedisKeyName.SI_KONG_OSD_HANGAR_ONLINE + ":*";
        List<String> droneKeys = redisProvider.getAllKeys(dronePattern).stream().map(ele -> ele.substring(ele.lastIndexOf(":") + 1)).collect(Collectors.toList());
        List<String> hangarKeys = redisProvider.getAllKeys(hangarPattern).stream().map(ele -> ele.substring(ele.lastIndexOf(":") + 1)).collect(Collectors.toList());
        Set<String> onLines = new HashSet<>(droneKeys);
        onLines.addAll(hangarKeys);
        for (String inLiveSn : inLiveKeys) {
            //  没有在上报中的直播关闭
            if (!onLines.contains(inLiveSn)) {
                siKongApiClientService.stopLiveByInLiveKey(inLiveSn);
            }
        }
    }

    /**
     * 上报消息处理
     */
    protected void handleLiveReportMessages(String message) {
        try {
            int active = 30;
            if (!message.contains(SiKongMessageType.live_hangar.name())) {
                return;
            }
            // 转换为实体
            WSMessageResultDTO wsMessageResultDTO = JSON.parseObject(message, WSMessageResultDTO.class);
            Map<String, Object> data = (Map<String, Object>) wsMessageResultDTO.getData();
            if (data == null) {
                return;
            }
            String hangarSn = SiKongMapDataUtil.getMapData("hangarSn", data);
            if (StringUtils.isNotEmpty(hangarSn)) {
                redisProvider.set(RedisKeyName.SI_KONG_OSD_REPORT_VIEW + ":" + hangarSn, "1", active);
            }
            String droneSn = SiKongMapDataUtil.getMapData("droneSn", data);
            if (StringUtils.isNotEmpty(hangarSn)) {
                redisProvider.set(RedisKeyName.SI_KONG_OSD_REPORT_VIEW + ":" + droneSn, "1", active);
            }
        } catch (Exception e) {
            log.error("处理直播上报信息错误", e);
        }
    }

    /**
     * 初始化定时器
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        initScheduledExecutorService();
    }
}
